package cas.ihep.hmss.hdfsin;

import cas.ihep.util.RefreshFileSize;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.globus.ftp.*;
import org.globus.ftp.exception.ServerException;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.lang.reflect.InvocationTargetException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * Created by chocolate on 2018/1/4.
 */
public class ApiMapper extends InMapper {
    private GridFTPClient client;
    private long totalsize=0;
    private long totaltime=0;

    public void run(Context context)  {
        try {
            client = new GridFTPClient("helion02.ihep.ac.cn", 5555);
//            Class<?> klass = GridFTPClient.class;
//            Field gLocalServer_field = klass.getDeclaredField("gLocalServer");
//            gLocalServer_field.setAccessible(true);
//            GridFTPServerFacade gLocalServer = (GridFTPServerFacade) gLocalServer_field.get(client);
//            gLocalServer.setOptions(new RetrieveOptions(16));
            client.authorize("xingg", "123456");
            client.setType(GridFTPSession.TYPE_IMAGE);
            client.setMode(GridFTPSession.MODE_EBLOCK);
            client.setDataChannelAuthentication(DataChannelAuthentication.NONE);
            client.setDataChannelProtection(1);
            client.setOptions(new RetrieveOptions(3));
            super.run(context);
            client.close(true);
        }catch (ServerException |IOException|InterruptedException e){
            e.printStackTrace();
            System.out.println("throw an exception");
        } finally {
            if (totaltime>0) {
                System.out.println("Speed: " + (totalsize / (totaltime / 1000.) / 1024 / 1024) + "MB/s");
            }else{
                System.out.println("Speed: 0MB/s");
            }
        }
    }

    protected void map(LongWritable lno, Text line, Context context) throws IOException, InterruptedException {
        try {
            String attemp = context.getTaskAttemptID().toString();
            LocatedBlock lb = null;
            HdfsFileStatus hfstatus = null;
            String opath = "";
            RefreshFileSize rfs = null;
            String path = line.toString();/*输入文件里的每一行*/
            Configuration conf = context.getConfiguration();
            String nn = conf.get("fs.default.name", "hdfs://192.168.60.41:8020");/*namenode*/
            String targetdir = conf.get("target.directory");
            URI uri;
            try {
                uri = new URI(nn);

            } catch (URISyntaxException e) {
                throw new IllegalArgumentException(e);
            }
            dfs = new DFSClient(uri, conf);
            //s4h = new Service4Hep(dfs,conf);
            s4h = constructor.newInstance(dfs, conf);
            rfs = new RefreshFileSize(dfs);
            int lastSlash = path.lastIndexOf('/');/*获取文件名*/
            String dest = targetdir + '/' + path.substring(lastSlash + 1);/*完整的目标逻辑路径*/
            System.out.println(dest);
            //hfstatus = s4h.createFile(dest, true, (short) 1, 2147483648L);/*新建文件获取元数据信息*/
            hfstatus = (HdfsFileStatus) createFile.invoke(s4h, dest, true, (short) 1, 2147483648L);
            //lb = s4h.addBlock(dest, hfstatus.getFileId());/*为文件分配block*/
            //String tmp = s4h.getTmpFile(lb);
            lb=(LocatedBlock) addBlock.invoke(s4h,dest,hfstatus.getFileId());
            String tmp=(String)getTmpFile.invoke(s4h,lb);
            opath = tmp + "/" + lb.getBlock().getBlockName();/*实际的物理路径*/
            System.out.println(opath);
            rfs.put(dest, hfstatus.getFileId(), opath);
            try {

                long size = client.getSize(path);
                System.out.println(size);
                totalsize += size;
                DataSink sink =
                        new FileRandomIO(new RandomAccessFile(opath, "rw"));
                System.out.println("==================");
                long start = System.currentTimeMillis();
                client.extendedGet(path, size, sink, null);
                totaltime += System.currentTimeMillis() - start;
                System.out.println("*************************");

            } catch (Exception e) {
                e.printStackTrace();
            }

            rfs.remove(dest);
            File f = new File(opath);
            if (f.exists()) {
                //s4h.calculatefileCheckSums(opath, lb);
                calculatefileCheckSums.invoke(s4h,opath,lb);
                lb.getBlock().setNumBytes(f.length());
                //s4h.close(lb, dest, hfstatus.getFileId());
                close0.invoke(s4h,lb,dest,hfstatus.getFileId());
            } else {
                dfs.delete(dest, true);
            }
            rfs.close();
            //s4h.close();
            close1.invoke(s4h);
            dfs.close();
            System.out.println("!!!!!!!!!!!");
        }catch (IllegalAccessException | InstantiationException e){
            e.printStackTrace();
            throw new IOException(e);
        }catch (InvocationTargetException e){
            e.getCause().printStackTrace();
            throw new IOException(e.getCause());
        }
    }

}

